#include <stdio.h>
#include <strings.h>
#include <string.h>
#include <stddef.h>
#include <stdlib.h>
#include <unistd.h>
#include <assert.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
//#include <ctype.h>

#include "MapReduceScheduler.h"
#include "stddefines.h"
#include "mapreduce_sort.h"
#include "wordcount_mapred.h"


int main(int argc, char *argv[]) {
   
   
   final_data_t wc_vals;
   int i;
   int fd;
   char * fdata;
   int disp_num;
   struct stat finfo;
   char * fname, * disp_num_str;

   struct timeval starttime,endtime;

   // Make sure a filename is specified
   if (argv[1] == NULL)
   {
      printf("USAGE: %s <filename> [Top # of results to display]\n", argv[0]);
      exit(1);
   }
   
   fname = argv[1];
   disp_num_str = argv[2];

   printf("Wordcount: Running...\n");
   
   // Read in the file
   CHECK_ERROR((fd = open(fname, O_RDONLY)) < 0);
   // Get the file info (for file length)
   CHECK_ERROR(fstat(fd, &finfo) < 0);
   // Memory map the file
   CHECK_ERROR((fdata = mmap(0, finfo.st_size + 1, 
      PROT_READ | PROT_WRITE, MAP_PRIVATE, fd, 0)) == NULL);
   
   // Get the number of results to display
   CHECK_ERROR((disp_num = (disp_num_str == NULL) ? 
      20 : atoi(disp_num_str)) <= 0);
   
   // Setup splitter args
   wc_data_t wc_data;
   wc_data.unit_size = 5; // approx 5 bytes per word
   wc_data.fpos = 0;
   wc_data.flen = finfo.st_size;
   wc_data.fdata = fdata;

   // Setup scheduler args
   scheduler_args_t sched_args;
   memset(&sched_args, 0, sizeof(scheduler_args_t));
   sched_args.task_data = &wc_data;
   sched_args.map = wordcount_map;
   sched_args.reduce = wordcount_reduce;
   sched_args.splitter = wordcount_splitter;
   sched_args.key_cmp = mystrcmp;
   sched_args.unit_size = wc_data.unit_size;
   sched_args.partition = NULL; // use default
   sched_args.result = &wc_vals;
   sched_args.data_size = finfo.st_size;
   sched_args.L1_cache_size = 32*1024;//atoi(GETENV("MR_L1CACHESIZE")); //1024 * 64;
   sched_args.num_map_threads = 8; //atoi(GETENV("MR_NUMTHREADS"));//8;
   sched_args.num_reduce_threads = 4; //atoi(GETENV("MR_NUMTHREADS"));//16;
   sched_args.num_merge_threads = 2; //atoi(GETENV("MR_NUMTHREADS")) / 2;//8;
   sched_args.num_procs = 4;//atoi(GETENV("MR_NUMPROCS"));//16;
   sched_args.key_match_factor = 2;//(float)atof(GETENV("MR_KEYMATCHFACTOR"));//2;

   printf("Wordcount: Calling MapReduce Scheduler Wordcount\n");

   gettimeofday(&starttime,0);

   CHECK_ERROR(map_reduce_scheduler(&sched_args) < 0);

   gettimeofday(&endtime,0);

   printf("Word Count: Completed %ld\n",(endtime.tv_sec - starttime.tv_sec));

   printf("Wordcount: MapReduce Completed\n");
   
   printf("Wordcount: Calling MapReduce Scheduler Sort\n");
   
   mapreduce_sort(wc_vals.data, wc_vals.length, sizeof(keyval_t), mykeyvalcmp);

   printf("Wordcount: MapReduce Completed\n");

   printf("\nWordcount: Results (TOP %d):\n", disp_num);
   for (i = 0; i < disp_num && i < wc_vals.length; i++)
   {
      keyval_t * curr = &((keyval_t *)wc_vals.data)[i];
      printf("%15s - %d\n", (char *)curr->key, (int)curr->val);
   }

   free(wc_vals.data);

   CHECK_ERROR(munmap(fdata, finfo.st_size + 1) < 0);
   CHECK_ERROR(close(fd) < 0);

   return 0;
}
